/**
 * 
 */
package com.rrd.dw.mr.text;

import java.io.IOException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.DocumentHelper;
import org.dom4j.Element;
import org.dom4j.Node;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rrd.dw.utils.HadoopUtils;

/**
 * @author xiexihao
 */
public class RiskAnalysisReportXMLParser extends Configured implements Tool {
    private static final String STR_FILED_SPLIT_FLAG = "\001";

    private static final String STR_TARGET_POS_BASIC = "basic";		//首贷基本表
    private static final String STR_TARGET_POS_LOAN_APP = "loanapp";	//首续贷共用表
    private static final String STR_TARGET_POS_ANTI_FRAUD_MODELS = "antifraudmodels";	//首续贷共用表
    private static final String STR_TARGET_POS_ONE_MONTH_CALL_INFO = "onemonthcallinfo";	//首续贷共用表
    private static final String STR_TARGET_POS_THREE_MONTH_CALL_INFO = "threemonthcallinfo";	//首续贷共用表

    protected static Logger log = LoggerFactory
        .getLogger(RiskAnalysisReportXMLParser.class);

    public static class AnalysisReportMapper
            extends Mapper<LongWritable, Text, Text, Text> {
        private Text key = null;
        private Text value = null;

        private Node tmpNode = null;
        private Document tmpDoc = null;

        protected void setup(Context context) {
            key = new Text();
            value = new Text();
        }

        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String[] columns = value.toString().split(STR_FILED_SPLIT_FLAG, -1);
            String id = columns[0];
            String report_key = columns[1];
            String source_system = columns[2];
            String user_key = columns[3];
            String xml = columns[4];
            String create_time = columns[5];
            String update_time = columns[6];
            String version = columns[7];

//			this.key.set(id);
            this.key.set(STR_TARGET_POS_BASIC);

            StringBuffer sb = new StringBuffer();
            sb.append(id).append(STR_FILED_SPLIT_FLAG);
            sb.append(report_key).append(STR_FILED_SPLIT_FLAG);
            sb.append(source_system).append(STR_FILED_SPLIT_FLAG);
            sb.append(user_key).append(STR_FILED_SPLIT_FLAG);

            try {
                tmpDoc = DocumentHelper.parseText(xml);
            } catch (DocumentException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }

            if (tmpDoc != null) {
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/top_out_10_mobile_percentage"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/top_in_10_mobile_percentage"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/face_plus_plus_thresholds/synthetic_face_threshold"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/face_plus_plus_thresholds/synthetic_face_confidence"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/six_month_living_city_call_percentage"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/six_month_lend_request_info/pass_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/six_month_lend_request_info/over_due_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/six_month_lend_request_info/need_repay_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/six_month_lend_request_info/apply_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/six_month_idnum_city_call_percentage"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/face_plus_plus_thresholds/screen_replay_threshold"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/face_plus_plus_thresholds/screen_replay_confidence"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/same_company_info/pass_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/same_company_info/over_due_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/same_company_info/need_repay_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/same_company_info/apply_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/salary"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/rule_id_by_tongdun"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_tongdun/final_score"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_tongdun/final_decision"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/reserved_phone_lend_request_info/pass_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/reserved_phone_lend_request_info/over_due_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/reserved_phone_lend_request_info/need_repay_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/reserved_phone_lend_request_info/apply_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/reserved_phone_call_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/register_phone_lend_request_info/pass_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/register_phone_lend_request_info/over_due_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/register_phone_lend_request_info/need_repay_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/register_phone_lend_request_info/apply_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/register_phone_call_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_qianhai_risk/qianhai_risk_total"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_qianhai_risk/qianhai_risk_score"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/profession"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/position"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/phone_register_location_code"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/phone_register_location_check"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/phone_register_location"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/phone_register"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/phone_package_type"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/phone_number_all_need_repay_amount"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/phone_in_use_time"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/phone_balance"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/outbound_call_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/os_platform"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/other_info/plist_info/online_shopping_app_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/dawn_day_night_call_percentage/night_percentage"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/nearly100_incoming_call_percentage"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_miguan/miguan_searched_total_by_idnum"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_miguan/miguan_searched_90days_by_idnum"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_miguan/miguan_searched_360days_by_idnum"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_miguan/miguan_searched_30days_by_idnum"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_miguan/miguan_searched_180days_by_idnum"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_miguan/miguan_gray_score"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_miguan/miguan_contacts_class2_blacklist_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_miguan/miguan_contacts_class1_blacklist_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/face_plus_plus_thresholds/mask_threshold"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/face_plus_plus_thresholds/mask_confidence"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/marriage"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/macao_call_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/loan_info/longitude"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/loan_info/latitude"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/loan_info/duration"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/loan_info/amount"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/living_time"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/other_info/plist_info/little_loan_app_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/incoming_outcoming_info/outcoming_time"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/incoming_outcoming_info/outcoming_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/incoming_outcoming_info/incoming_time"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/incoming_outcoming_info/incoming_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/id_number_six_prefix"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/id_number_location"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/id_number_lend_request_info/pass_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/id_number_lend_request_info/over_due_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/id_number_lend_request_info/need_repay_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/id_number_lend_request_info/apply_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_5km_lend_request_info/pass_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_5km_lend_request_info/over_due_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_5km_lend_request_info/need_repay_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_5km_lend_request_info/apply_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_3km_lend_request_info/pass_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_3km_lend_request_info/over_due_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_3km_lend_request_info/need_repay_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_3km_lend_request_info/apply_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_1km_lend_request_info/pass_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_1km_lend_request_info/over_due_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_1km_lend_request_info/need_repay_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/location_relativity_analysis/gps_nearby_1km_lend_request_info/apply_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/gender"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/face_plus_plus_thresholds/face_replaced"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/face_plus_plus_confidence"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/emergency_contact_call_number_rank_list/emergency_contact_call_number_rank/rank_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/face_plus_plus_thresholds/e_5"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/face_plus_plus_thresholds/e_4"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/face_plus_plus_thresholds/e_3"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_cross_validation/distance_living_company_address"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/device_lend_request_info/pass_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/device_lend_request_info/over_due_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/device_lend_request_info/need_repay_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/id_number_phone_number_history_statistics/device_lend_request_info/apply_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/dawn_day_night_call_percentage/day_percentage"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/dawn_day_night_call_percentage/dawn_percentage"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/contact_num"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/contact_lend_request_info/pass_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/contact_lend_request_info/over_due_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/contact_lend_request_info/need_repay_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/communication_statistics/contact_lend_request_info/apply_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/other_info/plist_info/common_social_app_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/collection_phone_incoming_info/phone_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/collection_phone_incoming_info/call_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/collection_phone_call_info/outcoming_time"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/collection_phone_call_info/outcoming_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/collection_phone_call_info/incoming_time"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/collection_phone_call_info/incoming_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/check_three_day_no_call"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/check_seven_day_no_call"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/check_salary"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/other_info/contact_lend_request_relativity/check_reserved_phone_in_contact_book"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/other_info/check_phone_same/check_register_taobao_phone_same"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/other_info/check_phone_same/check_register_reserved_phone_same"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/other_info/contact_lend_request_relativity/check_register_phone_in_contact_book"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_cross_validation/check_phone_id_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/check_name_id_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/check_living_company_city_same"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/check_living_address_pinpoint"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/other_info/check_phone_same/check_idnum_register_phone_match"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_cross_validation/check_gps_5km_living_address"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_cross_validation/check_gps_5km_company_address"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_cross_validation/check_gps_3km_living_address"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_cross_validation/check_gps_3km_company_address"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_cross_validation/check_gps_1km_living_address"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_cross_validation/check_gps_1km_company_address"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/check_fourteen_day_no_call"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/check_face_plus_plus_alive"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/check_emergency_contact_not_keep_in_touch"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/check_consecutive_phone_register"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/check_company_name_address"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/check_company_address_pinpoint"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/check_call_120"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/check_call_110"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_third_party_verification/check_bank_four_elements"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/age"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/tag_phone_analysis/n4/avg_total_time"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/tag_phone_analysis/n4/avg_out_time"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/tag_phone_analysis/n3/out_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/user_behavior_analysis/call_analysis/tag_phone_analysis/n3/avg_out_time"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/other_info/plist_info/loan_app_number"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_zhima_antifraud/verify_code"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_zhima_antifraud/hit"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_zhima_antifraud/score"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_zhima_antifraud/risk_code"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_tianji_xigua/decision_score"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/historial_relativity_analysis/check_black_list/result_by_zhima_credit_score"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(
                    "/feature/lend_request_analysis/lend_request_original_info/loan_info/loan_time"))
                    .append(STR_FILED_SPLIT_FLAG);
            }

            sb.append(update_time).append(STR_FILED_SPLIT_FLAG);
            sb.append(create_time).append(STR_FILED_SPLIT_FLAG);
            sb.append(version);

            this.value.set(sb.toString());
            context.write(this.key, this.value);

            /**
             * 解析首贷部分数据
             */
            parseLoanApp(context, id);
            parseOneMonthCallInfo(context, id);
            parseThreeMonthCallInfo(context, id);
            parseAntiFraudModels(context, id);

            /**
             * 解析续贷部分数据
             */

            tmpDoc = null;
        }

        private void parseLoanApp(Context context, String id)
                throws IOException, InterruptedException {
            this.key.set(STR_TARGET_POS_LOAN_APP);

            StringBuffer sb = new StringBuffer();

            tmpNode = tmpDoc.selectSingleNode(
                "/feature/other_info/plist_info/loan_app_list");
            if (tmpNode != null) {
                List<?> nds = tmpNode.selectNodes("loan_app");
                for (int i = 0; i < nds.size(); i++) {
                    Element app = (Element) nds.get(i);
                    Element appName = app.element("app_name");
                    Element packageName = app.element("package_name");
                    Element launchTime = app.element("launch_time");
                    Element lastUpdateTime = app.element("last_update_time");
                    Element appType = app.element("app_type");

                    sb.delete(0, sb.length());
                    sb.append(id).append(STR_FILED_SPLIT_FLAG);
                    sb.append(i + 1).append(STR_FILED_SPLIT_FLAG);
                    sb.append(appName == null ? "" : appName.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(packageName == null ? "" : packageName.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(launchTime == null ? "" : launchTime.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(
                        lastUpdateTime == null ? "" : lastUpdateTime.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(appType == null ? "" : appType.getText());

                    this.value.set(sb.toString());
                    context.write(this.key, this.value);
                }
            }
        }

        private void parseOneMonthCallInfo(Context context, String id)
                throws IOException, InterruptedException {
            this.key.set(STR_TARGET_POS_ONE_MONTH_CALL_INFO);

            StringBuffer sb = new StringBuffer();

            tmpNode = tmpDoc.selectSingleNode(
                "/feature/user_behavior_analysis/call_analysis/one_month_loan_app_customer_phone_call_info_list");
            if (tmpNode != null) {
                List<?> nds = tmpNode
                    .selectNodes("loan_app_customer_phone_call_info");
                for (int i = 0; i < nds.size(); i++) {
                    Element e = (Element) nds.get(i);
                    Element appName = e.element("app_name");
                    Element packageName = e.element("package_name");
                    Element incomingCount = e.element("incoming_number");
                    Element incomingTime = e.element("incoming_time");
                    Element outcomingCount = e.element("incoming_time");
                    Element outcomingTime = e.element("outcoming_time");

                    sb.delete(0, sb.length());
                    sb.append(id).append(STR_FILED_SPLIT_FLAG);
                    sb.append(i + 1).append(STR_FILED_SPLIT_FLAG);
                    sb.append(appName == null ? "" : appName.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(packageName == null ? "" : packageName.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(
                        incomingCount == null ? "" : incomingCount.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(
                        incomingTime == null ? "" : incomingTime.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(
                        outcomingCount == null ? "" : outcomingCount.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(
                        outcomingTime == null ? "" : outcomingTime.getText());

                    this.value.set(sb.toString());
                    context.write(this.key, this.value);
                }
            }
        }

        private void parseThreeMonthCallInfo(Context context, String id)
                throws IOException, InterruptedException {
            this.key.set(STR_TARGET_POS_THREE_MONTH_CALL_INFO);

            StringBuffer sb = new StringBuffer();

            tmpNode = tmpDoc.selectSingleNode(
                "/feature/user_behavior_analysis/call_analysis/three_month_loan_app_customer_phone_call_info_list");
            if (tmpNode != null) {
                List<?> nds = tmpNode
                    .selectNodes("loan_app_customer_phone_call_info");
                for (int i = 0; i < nds.size(); i++) {
                    Element e = (Element) nds.get(i);
                    Element appName = e.element("app_name");
                    Element packageName = e.element("package_name");
                    Element incomingCount = e.element("incoming_number");
                    Element incomingTime = e.element("incoming_time");
                    Element outcomingCount = e.element("incoming_time");
                    Element outcomingTime = e.element("outcoming_time");

                    sb.delete(0, sb.length());
                    sb.append(id).append(STR_FILED_SPLIT_FLAG);
                    sb.append(i + 1).append(STR_FILED_SPLIT_FLAG);
                    sb.append(appName == null ? "" : appName.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(packageName == null ? "" : packageName.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(
                        incomingCount == null ? "" : incomingCount.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(
                        incomingTime == null ? "" : incomingTime.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(
                        outcomingCount == null ? "" : outcomingCount.getText())
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(
                        outcomingTime == null ? "" : outcomingTime.getText());

                    this.value.set(sb.toString());
                    context.write(this.key, this.value);
                }
            }
        }

        private void parseAntiFraudModels(Context context, String id)
                throws IOException, InterruptedException {
            this.key.set(STR_TARGET_POS_ANTI_FRAUD_MODELS);

            StringBuffer sb = new StringBuffer();

            tmpNode = tmpDoc.selectSingleNode("/feature/anti_fraud_analysis");
            if (tmpNode == null)
                return;

            List<?> nds = tmpNode.selectNodes("./*");
            for (int i = 0; i < nds.size(); i++) {
                Element e = (Element) nds.get(i);
                if (!e.getName().toLowerCase().startsWith("model"))
                    continue;

                sb.delete(0, sb.length());
                sb.append(id).append(STR_FILED_SPLIT_FLAG);
                sb.append(e.getName()).append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(e.getPath() + "/is_success"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(e.getPath() + "/result/model_id"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(e.getPath() + "/result/loan_id"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(parseThroughXPath(e.getPath() + "/result/prob"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(
                    parseThroughXPath(e.getPath() + "/result/model_feature"));

                this.value.set(sb.toString());
                context.write(this.key, this.value);
            }
        }

        private String parseThroughXPath(String xpath) {
            tmpNode = tmpDoc.selectSingleNode(xpath);
            return tmpNode == null ? "" : tmpNode.getStringValue();
        }
    }

    public static class AnalysisReportReducer
            extends Reducer<Text, Text, Text, Text> {
        private NullWritable nullkey = NullWritable.get();
        private MultipleOutputs<Text, Text> mos = null;

        protected void setup(Context context)
                throws IOException, InterruptedException {
            mos = new MultipleOutputs<Text, Text>(context);
        }

        protected void cleanup(Context context)
                throws IOException, InterruptedException {
            mos.close();
        }

        public void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
//			for(Text t : values)
//				context.write(key2, new Text(key.toString() + STR_FILED_SPLIT_FLAG + t.toString()));

            String keyStr = null;

            for (Text t : values) {
                keyStr = key.toString();
                if (STR_TARGET_POS_BASIC.equals(keyStr)) {
                    this.mos.write(STR_TARGET_POS_BASIC, nullkey, t);
                } else if (STR_TARGET_POS_LOAN_APP.equals(keyStr)) {
                    this.mos.write(STR_TARGET_POS_LOAN_APP, nullkey, t);
                } else if (STR_TARGET_POS_ANTI_FRAUD_MODELS.equals(keyStr)) {
                    this.mos.write(STR_TARGET_POS_ANTI_FRAUD_MODELS, nullkey,
                        t);
                } else if (STR_TARGET_POS_ONE_MONTH_CALL_INFO.equals(keyStr)) {
                    this.mos.write(STR_TARGET_POS_ONE_MONTH_CALL_INFO, nullkey,
                        t);
                } else if (STR_TARGET_POS_THREE_MONTH_CALL_INFO
                    .equals(keyStr)) {
                    this.mos.write(STR_TARGET_POS_THREE_MONTH_CALL_INFO,
                        nullkey, t);
                }
            }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        // TODO Auto-generated method stub
        Configuration conf = getConf();
        Job job = Job.getInstance(conf);

        System.setProperty("HADOOP_USER_NAME", "hdfs");
        String dt = conf.get("etl.param.dt", "2017-07-27");
        String inputDir = "/data/bdm/pdl03_ra_report_inc/dt=" + dt;
        String outputDir = conf.get("etl.param.output",
            "/tmp/fdm/verifyanalysisreport");
        if (dt.equals("")) {
            log.info("请输入日期：etl.param.dt");
            return -1;
        }
        if (outputDir.equals("")) {
            log.info("请输入输出路径：etl.param.output");
            return -1;
        }

        log.info("输入路径:" + inputDir);
        log.info("输出路径:" + outputDir);

        HadoopUtils.delete(conf, new Path(outputDir));

        job.setJarByClass(this.getClass());

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setMapperClass(AnalysisReportMapper.class);
        job.setReducerClass(AnalysisReportReducer.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        FileInputFormat.setInputPaths(job, new Path(inputDir));
        FileOutputFormat.setOutputPath(job, new Path(outputDir));

        MultipleOutputs.addNamedOutput(job, STR_TARGET_POS_BASIC,
            TextOutputFormat.class, Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, STR_TARGET_POS_LOAN_APP,
            TextOutputFormat.class, Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, STR_TARGET_POS_ANTI_FRAUD_MODELS,
            TextOutputFormat.class, Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, STR_TARGET_POS_ONE_MONTH_CALL_INFO,
            TextOutputFormat.class, Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job,
            STR_TARGET_POS_THREE_MONTH_CALL_INFO, TextOutputFormat.class,
            Text.class, Text.class);

        return job.waitForCompletion(true) ? 0 : 1;
    }

    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
        System.exit(ToolRunner.run(new RiskAnalysisReportXMLParser(), args));
    }
}
